package com.atguigu.flink0922.chapter11;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/12 11:09
 */
public class Flink13_Function_Table {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        DataStreamSource<String> stream = env.fromElements("hello,atguigu,world", "aaa,bbbbb", "");
        
        Table table = tEnv.fromDataStream(stream, $("line"));
        
        // 1. 内联使用
        table
            .joinLateral(call(Split.class, $("line")))
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();
        
        table
            .leftOuterJoinLateral(call(Split.class, $("line")))
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();
        
        // 2. 注册后使用
        tEnv.createTemporaryFunction("split", Split.class);
        table
            .joinLateral(call("split", $("line")))
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();
    }
    
    @FunctionHint(output = @DataTypeHint("ROW(word string, len int)"))
    public static class Split extends TableFunction<Row> {
        public void eval(String line) {
            if (line.length() == 0) {
                return;
            }
            for (String s : line.split(",")) {
                // 来一个字符串, 按照逗号分割, 得到多行, 每行为这个单词和他的长度
                collect(Row.of(s, s.length()));
            }
        }
    }
    
}



